package com.example.example;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class UserRetentionAnalysis {

    /**
     * 程序入口函数
     * <p>
     * 主要功能包括：
     * 1. 初始化SparkSession并启用Hive支持；
     * 2. 执行SQL查询以计算用户在1天、3天、5天后的留存率；
     * 3. 将分析结果写入MySQL数据库中的指定表；
     * 4. 关闭SparkSession资源。
     * </p>
     *
     * @param args 命令行参数（未使用）
     */
    public static void main(String[] args) {

        SparkSession spark = SparkSession.builder().appName("UserRetentionAnalysis")
                .master("local[*]")
                .enableHiveSupport()
                .getOrCreate();

        // 执行用户留存分析SQL查询
        // 子查询部分统计每个日期的新增用户数以及后续第1、3、5天的回访用户数
        // 外层查询计算对应的留存率并格式化为百分比形式

        Dataset<Row> ds = spark.sql("SELECT " +
                "       day, " +
                "       CAST((reten_1/reten_0)*100 AS DECIMAL(10,2)) AS reten_1_rate, " +
                "       CAST((reten_3/reten_0)*100 AS DECIMAL(10,2)) AS reten_3_rate, " +
                "       CAST((reten_5/reten_0)*100 AS DECIMAL(10,2)) AS reten_5_rate " +
                "FROM ( " +
                "   SELECT " +
                "       a.day, " +
                "       count(distinct b.user_id) as reten_0, " +
                "       count(distinct if(datediff(b.day,a.day)=1,b.user_id,null)) as reten_1, " +
                "       count(distinct if(datediff(b.day,a.day)=3,b.user_id,null)) as reten_3, " +
                "       count(distinct if(datediff(b.day,a.day)=5,b.user_id,null)) as reten_5 " +
                "   FROM " +
                "       (SELECT user_id,day FROM dwd.dwd_behavior_info GROUP BY user_id,day) a " +
                "   JOIN    (SELECT user_id,day FROM dwd.dwd_behavior_info GROUP BY user_id,day) b " +
                "   ON a.user_id = b.user_id " +
                "   WHERE a.day <= b.day " +
                "   GROUP BY a.day " +
                ") t");

        ds.write()
                .format("jdbc")
                .option("url","jdbc:mysql://hadoop:3306/ana_behavior_db")
                .option("dbtable","user_retention_analysis")
                .option("user","root")
                .option("password","123456")
                .option("driver","com.mysql.cj.jdbc.Driver")
                .mode("overwrite")
                .save();

        spark.close();
    }
}
